/*
 * Copyright (c) 2018-2023 北京华恒盛世科技有限公司
 * QUICKPOOL is licensed under Mulan PSL v2.
 * You can use this software according to the terms and conditions of the Mulan PSL v2.
 * You may obtain a copy of Mulan PSL v2 at:
 *     http://license.coscl.org.cn/MulanPSL2
 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
 * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
 * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
 * See the Mulan PSL v2 for more details.
*/

#include "tclinfo.h"
#include "jm.h"

pid_t gpidUnrm = -1;
int gboolDefaultUnrm = FALSE;

#if defined (HAVE_HWLOC_H)
hwloc_topology_t topology;
#endif

static time_t stimeCheckUnrm;
static FILE *spipeUnrm;
static numa_obj_t *sptrMyNuma;


static struct cluster *set_mycluster(char *, char *);
static void set_cluster_resmap(int, int *, int *);

static struct hostNode *hostinfo_2node(struct QuickPoolHostInfo *);
static void location_2resinst(struct resourceLoc *);
static void reorder_resource(void);

static void get_unrm_report(void);

#if defined (HAVE_HWLOC_H)
static void get_numa_info(void);
#endif

/**
 * @brief       As node resource manager, initialize myself.
 *
 * @retval      0            #1: succeed
 * @retval      js_die       #2: die if cannot do initialization
 */
int
init_nrm_info(bool_t confFlag)
{
    gptrMyLoad = (struct loadIndex *)calloc(NUMBUILTIN, sizeof(struct loadIndex));
    if (NULL == gptrMyLoad) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", NUMBUILTIN*sizeof(struct loadIndex));
        jm_die(DAEMON_QUIT_MEM);
    }

    gptrMyLoad[R1M].increasing = TRUE;
    gptrMyLoad[R1M].delta[0] = 0.30;
    gptrMyLoad[R1M].delta[1] = 0.10;
    gptrMyLoad[R1M].sigdiff = 0.10;

    gptrMyLoad[R5M].increasing = TRUE;
    gptrMyLoad[R5M].delta[0] = 0.15;
    gptrMyLoad[R5M].delta[1] = 0.10;
    gptrMyLoad[R5M].sigdiff = 0.10;

    gptrMyLoad[R15M].increasing = TRUE;
    gptrMyLoad[R15M].delta[0] = 0.15;
    gptrMyLoad[R15M].delta[1] = 0.10;
    gptrMyLoad[R15M].sigdiff = 0.10;

    gptrMyLoad[UT].increasing = TRUE;
    gptrMyLoad[UT].delta[0] = 1.00;
    gptrMyLoad[UT].delta[1] = 1.00;
    gptrMyLoad[UT].sigdiff = 0.10;

    gptrMyLoad[PG].increasing = TRUE;
    gptrMyLoad[PG].delta[0] = 2.5;
    gptrMyLoad[PG].delta[1] = 1.5;
    gptrMyLoad[PG].sigdiff = 5.0;

    gptrMyLoad[IO].increasing = TRUE;
    gptrMyLoad[IO].delta[0] = 80;
    gptrMyLoad[IO].delta[1] = 40;
    gptrMyLoad[IO].sigdiff = 5.0;

    gptrMyLoad[SWP].increasing = FALSE;
    gptrMyLoad[SWP].delta[0] = 10;
    gptrMyLoad[SWP].delta[1] = 10;
    gptrMyLoad[SWP].sigdiff = 2.0;

    gptrMyLoad[MEM].increasing = FALSE;
    gptrMyLoad[MEM].delta[0] = 10;
    gptrMyLoad[MEM].delta[1] = 10;
    gptrMyLoad[MEM].sigdiff = 3.0;

    char confFile[PATH_MAX+1];
    snprintf(confFile, PATH_MAX, "%s/qp.shared", baseParams[BASE_CONFDIR].string_value);
    if (NULL == readconf_shared(confFile)) { // read qp.shared file

        if (quickpool_errno == ERROR_CONF_WARNING) {
            gintConfigError = WARNING_ERR;
        }else {
            jm_die(DAEMON_QUIT_CONFFILE);
        }
    }

    set_mycluster(gptrSharedConf->cluster_name, gptrSharedConf->servers);

    if (NULL == gptrMyCluster) {
        log_message(ERR, NOCLASS, "My cluster is not configured in qp.shared.");
        jm_die(DAEMON_QUIT_FATAL);
    }

    snprintf(confFile, PATH_MAX, "%s/qp.cluster.%s",
             baseParams[BASE_CONFDIR].string_value, gptrMyCluster->name);

    if (NULL == readconf_cluster(confFile)) { // read qp.cluster file
        if (quickpool_errno == ERROR_CONF_WARNING) {
            gintConfigError = WARNING_ERR;
        }else {
            jm_die(DAEMON_QUIT_CONFFILE);
        }
    }

    reorder_resource();

    if (initialize_parser_hash(gptrSharedConf->qpinfo) < 0) {
        jm_die(DAEMON_QUIT_MEM);
    }

    unsigned int i;
    for (i=0; i<gptrClusterConf->num_hosts; i++) {
        hostinfo_2node(&(gptrClusterConf->hosts[i])); // set node information
    }

    for (i=0; i<gptrClusterConf->num_location; i++) {
        location_2resinst(&(gptrClusterConf->loc[i])); // resource relationship with node
    }

    LIST_T *dynamicNodeList;
    dynamicNodeList = gf_list_make("JM dynamic node list");

    load_dynamic_node(dynamicNodeList);

    LIST_ITERATOR_T iter;
    gf_listiterator_link(&iter, dynamicNodeList);

    struct dynamicNode *dynNode;
    for (dynNode = (struct dynamicNode *)gf_listiterator_curentry(&iter);
         !gf_listiterator_isend(&iter); gf_listiterator_getnext(&iter, (LIST_T **)&dynNode)) {

        if (NULL == dynNode || NULL == dynNode->node_req) {
            continue;
        }

        struct request *nodeReq = dynNode->node_req;

        char *strNode=NULL;
        for (i=0; i<nodeReq->number; i++) {
            switch (nodeReq->keyvalues[i].key) {
            case REQUEST_KEY_HOSTNAME:
                strNode = nodeReq->keyvalues[i].value;
                break;
            default:
                break;
            }

            if (NULL != strNode) {
                break;
            }
        }

        struct hostNode *host=NULL;
        char message[MAXMESSAGELENGTH];
        host = add_hostnode(nodeReq, message);
        if (NULL == host) {
            log_message(ERR, NOCLASS, "%s: Add dynamic node <%s> failed due to %s",
                        __func__, strNode, message);
            continue;
        }

        host->server = DYNAMIC_SERVER;
    }
    gf_list_free(dynamicNodeList,free_dynamic_node);

    char *hname;
    if ((hname = qp_getmyhostname()) == NULL) {
        jm_die(DAEMON_QUIT_FATAL);
    }

    gptrMyHost = (struct hostNode *)gf_hash_find(gptrMyCluster->node_table, hname);
    if (NULL == gptrMyHost) {
        log_message(ERR, NOCLASS,"%s: In %s, master host %s is not configured in Host section of file qp.cluster.%s", 
                    __func__, confFile, hname, gptrMyCluster->name);

        jm_die(DAEMON_QUIT_FATAL);
    }

    for (i = 0; i < GET_INTNUM(gptrSharedConf->qpinfo->num_index) + 1; i++) {
        gptrMyHost->status[i] = 0;
    }

    gptrMyCluster->status = CLUST_STAT_OK | CLUST_ACTIVE | CLUST_INFO_AVAIL;
    gptrMyCluster->main_admin = gptrClusterConf->admin->admin_names[0];
    gptrMyCluster->main_adminid = gptrClusterConf->admin->admin_ids[0];

    gptrMyCluster->num_admins = gptrClusterConf->admin->num_admins;
    gptrMyCluster->adminids = gptrClusterConf->admin->admin_ids;
    gptrMyCluster->admins = gptrClusterConf->admin->admin_names;

    unsigned int numRes = GET_INTNUM(gptrSharedConf->qpinfo->num_resources);
    if (NULL == gptrMyCluster->res_bitmap && 0 < numRes) {

        gptrMyCluster->res_bitmap = (int *)calloc(numRes, sizeof(int));
        if (NULL == gptrMyCluster->res_bitmap) {

            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "calloc", numRes*sizeof(int));
            jm_die(DAEMON_QUIT_MEM);
        }
    }

    unsigned int numTypes = GET_INTNUM(gptrSharedConf->qpinfo->num_types);
    if (NULL == gptrMyCluster->type_bitmap && 0 < numTypes) {

        gptrMyCluster->type_bitmap = (int *)calloc(numTypes, sizeof(int));
        if (NULL == gptrMyCluster->type_bitmap) {

            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "calloc", numTypes*sizeof(int));
            jm_die(DAEMON_QUIT_MEM);
        }
    }

    unsigned int numModels = GET_INTNUM(gptrSharedConf->qpinfo->num_models);
    if (NULL == gptrMyCluster->model_bitmap && 0 < numModels) {

        gptrMyCluster->model_bitmap = (int *)calloc(numModels, sizeof(int));
        if (NULL == gptrMyCluster->model_bitmap) {

            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "calloc", numModels*sizeof(int));
            jm_die(DAEMON_QUIT_MEM);
        }
    }

    HASH_WALK_T walk;
    char *key;
    gf_hash_walk_start(gptrMyCluster->node_table, &walk);
    struct hostNode *host;
    while (NULL != (host = (struct hostNode *)gf_hash_walk(&walk, &key))) {
        if (0 <= host->type_index) {
            SET_BIT(host->type_index, gptrMyCluster->type_bitmap);
        }
        if (0 <= host->model_index) {
            SET_BIT(host->model_index, gptrMyCluster->model_bitmap);
        }

        set_cluster_resmap(gptrSharedConf->qpinfo->num_resources, gptrMyCluster->res_bitmap,
                           host->res_bitmap);
    }
    gf_hash_walk_end(&walk);

    if (0 > initialize_tcl(gptrSharedConf->qpinfo->num_index, gptrSharedConf->qpinfo->num_user_index,
                           gptrSharedConf->qpinfo->num_resources, gptrSharedConf->qpinfo->bool_start_index,
                           gptrSharedConf->qpinfo->string_start_index, gptrSharedConf->qpinfo->resource_table)) {
        jm_die(DAEMON_QUIT_FATAL);
    }

    set_static_info();

    gptrMyHost->noload_count = 0;

    if (NUMBUILTIN < gptrSharedConf->qpinfo->num_index) {

        gptrMyLoad = (struct loadIndex *)realloc(gptrMyLoad,
                                                 sizeof(struct loadIndex)*gptrSharedConf->qpinfo->num_index);
        if (NULL == gptrMyLoad) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "realloc", gptrSharedConf->qpinfo->num_index*sizeof(struct loadIndex));
            jm_die(DAEMON_QUIT_MEM);
        }

        for (i=NUMBUILTIN; i<gptrSharedConf->qpinfo->num_index; i++) {

            memset(&(gptrMyLoad[i]), 0, sizeof(struct loadIndex));
            gptrMyLoad[i].increasing = gptrSharedConf->qpinfo->resource_table[i].order_type == INCR;
            gptrMyLoad[i].delta[0] = 9000;
            gptrMyLoad[i].delta[1] = 9000;
            gptrMyLoad[i].sigdiff = 0.0001;
        }
    }

    for (i = 0; i < gptrSharedConf->qpinfo->num_index; i++) {
        gptrMyLoad[i].busy_value = gptrMyHost->busy_threshold[i];
    }

    return 0;
} // end function init_nrm_info

#if defined (HAVE_HWLOC_H)
void
free_numanode(numa_obj_t *node)
{
    if (NULL == node) {
        return;
    }

    numa_obj_t *child;
    child = node->child;
    for (; child != NULL; child = child->back) {
        free_numanode(child);
    }
    FREEUP(node);

    return;
} // end function free_numanode
#endif

struct hostNode *
create_hostnode(void)
{
    struct hostNode *hNode;

    hNode = (struct hostNode *)calloc(1, sizeof(struct hostNode));
    if (NULL == hNode) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", sizeof(struct hostNode));
        jm_die(DAEMON_QUIT_MEM);
    }

    hNode->res_bitmap = (int *)calloc(GET_INTNUM(gptrSharedConf->qpinfo->num_resources), sizeof(int));
    if (NULL == hNode->res_bitmap) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", GET_INTNUM(gptrSharedConf->qpinfo->num_resources)*sizeof(int));
        jm_die(DAEMON_QUIT_MEM);
    }

    hNode->status = (int *)calloc((1 + GET_INTNUM(gptrSharedConf->qpinfo->num_index)), sizeof(int));
    if (NULL == hNode->status) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", (1+GET_INTNUM(gptrSharedConf->qpinfo->num_resources))*sizeof(int));
        jm_die(DAEMON_QUIT_MEM);
    }

    hNode->load = (double *)calloc(gptrSharedConf->qpinfo->num_index, sizeof(double));
    if (NULL == hNode->load) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", gptrSharedConf->qpinfo->num_index*sizeof(double));
        jm_die(DAEMON_QUIT_MEM);
    }

    hNode->busy_threshold =  (double *)calloc(gptrSharedConf->qpinfo->num_index, sizeof(double));
    if (NULL == hNode->busy_threshold) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", gptrSharedConf->qpinfo->num_index*sizeof(double));
        jm_die(DAEMON_QUIT_MEM);
    }

    unsigned int i;
    for (i=0; i<gptrSharedConf->qpinfo->num_index; i++) {
        hNode->load[i] = INFINIT_LOAD;
    }

    for (i=NUMBUILTIN; i<gptrSharedConf->qpinfo->num_index; i++) {
        hNode->busy_threshold[i] = (gptrSharedConf->qpinfo->resource_table[i].order_type == INCR) ?
                                   INFINIT_LOAD : -INFINIT_LOAD;
    }

    hNode->status[0] = NRM_UNAVAIL;

    return hNode;
} // end function create_hostnode

void
free_hostnode(struct hostNode *host)
{
    if (NULL == host) {
        return;
    }

    FREEUP(host->name);
    FREEUP(host->addr);

    FREEUP(host->busy_threshold);
    FREEUP(host->load);
    FREEUP(host->res_bitmap);
    FREEUP(host->status);
    gf_hash_free(host->inst_table, gf_free_none);

    FREEUP(host);

    return;
}

int
get_hosttype_index(const char *name)
{
    unsigned int i;

    for (i = 0; i < gptrSharedConf->qpinfo->num_types; i++) {

        if (0 != strcmp(gptrSharedConf->qpinfo->host_types[i], name)) {
            continue;
        }

        return i;
    }

    return -1;
}

int
get_hostmodel_index(const char *modelname)
{
    unsigned int i;

    for (i = 0; i < gptrSharedConf->qpinfo->num_models; i++) {
        if (0 != strcmp(gptrSharedConf->qpinfo->host_models[i], modelname)) {
            continue;
        }

        return i;
    }

    return -1;
}

/**
 * @brief       Set my node static information.
 *
 * @note        Should call send_static_info() to send my static info out later.
 */
void
set_static_info(void)
{
    gptrMyHost->load[R1M] =  0.0;
    gptrMyHost->load[R5M]  =  0.0;
    gptrMyHost->load[R15M] =  0.0;

    double etime;
    double itime;

    get_cputime(&itime, &etime);

    get_memswap();
    gptrMyHost->static_info.maxmem = (double)get_maxmem() / 1024.0;
    gptrMyHost->static_info.maxswap = (double)get_maxswp() / 1024.0;

    struct statvfs fs;
    if (statvfs("/tmp", &fs) < 0) {
        log_message(ERR, NOCLASS, "%s: statvfs() failed /tmp: %m", __func__);
        gptrMyHost->static_info.maxtmp = 0.0;
    } else {
        gptrMyHost->static_info.maxtmp = (double)fs.f_blocks/((double)(1024 * 1024)/fs.f_bsize);
    }

    char *strNcpu;
    strNcpu = getenv("QP_NCPUS");
    if (NULL != strNcpu) {
        gptrMyHost->static_info.maxcpu = atoi(strNcpu);
    } else {
        gptrMyHost->static_info.maxcpu = get_cores();
    }

    if (0 >= gptrMyHost->static_info.maxcpu) {
        log_message(ERR, NOCLASS, "%s: got wrong cpu number [%d]. Set it to 1",
                    __func__, gptrMyHost->static_info.maxcpu);
        gptrMyHost->static_info.maxcpu = 1;
    }

    gptrMyHost->static_info.host_type = get_hosttype();
    int typeNo = get_hosttype_index(gptrMyHost->static_info.host_type);
    if (0 <= typeNo) {
        gptrMyHost->type_index = typeNo;
    } else {
        typeNo = add_hosttype(gptrMyHost->static_info.host_type);
        if (0 <= typeNo) {
            gptrMyHost->type_index = typeNo;
        }
    }

    gptrMyHost->static_info.host_arch = get_hostmodel();

    gptrMyHost->num_GPU = 0;
    get_gpu_info();

#if defined (HAVE_HWLOC_H)
    RECORD_FUNCTIME(0, get_numa_info(), "get_numa_info()");
#endif
    gptrMyHost->static_info.numa = sptrMyNuma;
    return;
} // end function set_static_info

/**
 * @brief       Set my load values and send it out to master.
 *
 * @note        For master, if do not receive load from slave for
 *              several times, mark the node unavail.
 */
void
getload_andsend(void)
{
    double cpuUsage;
    double r1m=0.0, r5m=0.0, r15m=0.0;

    RECORD_FUNCTIME(0, get_unrm_report(), "get_unrm_report()");

    get_system_load(&r1m, &r5m, &r15m);

    double etime=-1;
    double itime;
    cpuUsage = get_cputime(&itime, &etime);

    get_memswap();

    gptrMyHost->load[PG] = get_paging(etime);
    gptrMyHost->load[IO] = iorate(etime);
    gptrMyHost->load[MEM] = (double)get_freemem() / (double)1024.0;
    gptrMyHost->load[SWP] = (double)get_freeswap() / (double)1024.0;
    gptrMyHost->load[TMP] = (double)get_availtmp();
    gptrMyHost->load[R1M] = r1m;
    gptrMyHost->load[R5M]  = r5m;
    gptrMyHost->load[R15M] = r15m;
    gptrMyHost->load[UT] = cpuUsage;

#if defined(QUICKPOOL_WITH_CUDA)
    get_gpu_info();
#endif

    unsigned int i;
    for (i = 0; i < gptrSharedConf->qpinfo->num_index; i++) {

        gptrMyLoad[i].value = gptrMyHost->load[i];

        if ((gptrMyLoad[i].increasing && fabs(gptrMyLoad[i].value - INFINIT_LOAD) < 1.0)
            || (!gptrMyLoad[i].increasing && fabs(gptrMyLoad[i].value + INFINIT_LOAD) < 1.0)) {
            continue;
        }

        if (gptrMyLoad[i].increasing) {
            if (gptrMyLoad[i].value > gptrMyLoad[i].busy_value) {
                SET_BIT(i + INTEGER_BITS, gptrMyHost->status);
                gptrMyHost->status[0] |= NRM_BUSY;
            } else {
                CLEAR_BIT(i + INTEGER_BITS, gptrMyHost->status);
            }
        } else {
            if (gptrMyLoad[i].value < gptrMyLoad[i].busy_value) {
                SET_BIT(i + INTEGER_BITS, gptrMyHost->status);
                gptrMyHost->status[0] |= NRM_BUSY;
            } else {
                CLEAR_BIT(i + INTEGER_BITS, gptrMyHost->status);
            }
        }
    }

    int busyBits=0;
    for (i = 0; i < GET_INTNUM (gptrSharedConf->qpinfo->num_index); i++) {
        busyBits += gptrMyHost->status[i+1];
    }

    if (!busyBits) {
        gptrMyHost->status[0] &= ~NRM_BUSY;
    }

    if (gboolMaster) { // I am master, check slave load update and mark status.
        HASH_WALK_T walk;
        char *key;
        struct hostNode *host;

        gf_hash_walk_start(gptrMyCluster->node_table, &walk);
        while (NULL != (host=(struct hostNode *)gf_hash_walk(&walk, &key))) {

            if (host == gptrMyHost) {
                continue;
            }

            if (host->server == CLIENT) {
                continue;
            }

            host->noload_count ++;
            if (host->noload_count > 10000)
                host->noload_count = 100;

            if (STATUS_ISUNAVAIL(host->status)) {
                continue;
            }

            if (host->noload_count <= (gintInactivityLimit + gintInactiveRetryLimit)) {
                continue;
            }

            log_message(INFO, CHANNEL, "%s: node <%s> is unavailable, about <%d> round no load.",
                        __func__, host->name, host->noload_count);

            host->status[0] |= NRM_UNAVAIL;

            HASH_WALK_T walkNodeInst;
            char *keyNodeInst;
            struct resourceInstance *inst;
            gf_hash_walk_start(host->inst_table, &walkNodeInst);
            while (NULL != (inst=(struct resourceInstance *)gf_hash_walk(&walkNodeInst,
                                                                         &keyNodeInst))) {
                if (inst->update_host != host) {
                    continue;
                }

                if (inst->resource->flags & RESOURCE_FLAG_DYNAMIC) {
                    strcpy(inst->value, "-");
                }
            }
            gf_hash_walk_end(&walkNodeInst);
        }
        gf_hash_walk_end(&walk);
    }

    if (!gboolMaster) {
        RECORD_FUNCTIME(0, send_myload(), "send_myload()");
    }

    return;
} // end function getload_andsend

/**
 * @brief       Set flag for node who would start unrm.
 *
 * @note        Go through shared resource table to set call_unrm flag for node.
 *              For shared all resource, master node is the caller.
 */
void
set_unrm_caller(void)
{
    HASH_WALK_T walk;
    char *key;
    struct sharedResource *res;
    gf_hash_walk_start(ghashResource, &walk);
    while (NULL != (res=(struct sharedResource *)gf_hash_walk(&walk, &key))) {

        struct resourceInstance *inst;
        if (0 == LIST_NUM_ENTS(res->inst_list)) { // no dynamic shared resource.
            continue;
        }


        for (inst = (struct resourceInstance *)res->inst_list->forw;
             inst != (struct resourceInstance *)res->inst_list;
             inst = (struct resourceInstance *)inst->forw) {

            if (0 == HASH_NUM_ENTS(inst->node_table)) {
                continue;
            }

            if (gboolMaster && res->flags & RESOURCE_FLAG_SHARED_ALL) {
                inst->update_host = gptrMyHost;
            }

            if (inst->update_host == gptrMyHost) {
                gptrMyHost->call_unrm = TRUE;
                continue;
            }
        }
    }
    gf_hash_walk_end(&walk);

    return;
} // end function set_unrm_caller

void
smooth(double *val, double instant, double factor)
{
    (*val) = ((*val) * factor) + (instant * (1 - factor));
    return;
} /* smooth */


/* ------------------------------------------------
 *
 *  below are static functions used in this file
 *
 * ------------------------------------------------ */

/**
 * @brief       Create gptrMyCluster and set its members.
 *
 * @param[in]   clustername  #1: my cluster name
 * @param[in]   servers      #2: compute node of my cluster
 *
 * @retval      not NULL     #1: succeed
 * @retval      NULL         #2: failed, system call failure
 */
static struct cluster *
set_mycluster(char *clustername, char *servers)
{
    int i = 0;
    struct hostent *hp;

    if (NULL != gptrMyCluster) {
        log_message(ERR, NOCLASS, "%s: Ignoring duplicate cluster %s",
                    __func__, gptrSharedConf->cluster_name);
        return NULL;
    }

    gptrMyCluster = (struct cluster *)calloc(1, sizeof(struct cluster));
    if (NULL == gptrMyCluster) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", sizeof(struct cluster));
        jm_die(DAEMON_QUIT_MEM);
    }

    gptrMyCluster->name = copy_string(clustername);

    if (NULL != servers && '\0' != servers[0]) {
        char *sp;
        sp = servers;
        i = 0;

        unsigned int len = strlen(sp);
        char *word;
        word = (char *)calloc(len+1, sizeof(char));
        if (NULL == word) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", len+1);
            jm_die(DAEMON_QUIT_MEM);
        }

        char *token;
        while ((token = get_string_token(&sp, " ", word, len+1)) != NULL && i < MAXCANDHOSTS) {

            len = strlen(sp);
            hp = gf_gethostbyname(token);
            if (!hp) {
	        log_message(ERR, NOCLASS, "%s: Invalid host %s for cluster %s, ignoring",
	                    __func__, token, clustername);
                continue;
            }
        }

        FREEUP(word);
    }

    gptrMyCluster->status = CLUST_ACTIVE | CLUST_STAT_UNAVAIL;
    gptrMyCluster->number_inactivity = 0;

    gptrMyCluster->node_table = gf_hash_make(100);
    if (NULL == gptrMyCluster->node_table) {
        jm_die(DAEMON_QUIT_MEM);
    }

    return gptrMyCluster;
} // end function set_mycluster

static void 
set_cluster_resmap(int num, int *tobitmaps, int *frombitmaps)
{
    int j;
    for (j = 0; j < GET_INTNUM(num); j++) {
        tobitmaps[j] = (tobitmaps[j] | frombitmaps[j]);
    }

    return;
}

/**
 * @brief       Transfer QuickPoolHostInfo to hostNode.
 *
 * @param[in]   info         #1: host info read from configuration file
 *
 * @retval      not NULL     #1: succeed
 * @retval      NULL         #2: failed, system call failure
 * 
 * @note        We use node data in jm.
 */
static struct hostNode *
hostinfo_2node(struct QuickPoolHostInfo *info)
{
    struct hostNode *host;
    host = (struct hostNode *)gf_hash_find(gptrMyCluster->node_table, info->host);
    if (NULL != host) {
        log_message(WARNING, NOCLASS, "%s: host %s redefined, using previous definition",
                    __func__, info->host);
        return host;
    }

    if ((host = create_hostnode()) == NULL) {
        log_message(ERR, NOCLASS, "%s: create_hostnode() failed %m", __func__);
        return NULL;
    }

    if (NULL == info->host_model) {
        host->model_index = 0;
    } else if ((host->model_index = get_hostmodel_index(info->host_model)) < 0) {
        log_message(ERR, NOCLASS, "%s: Unknown host model %s for host <%s>.",
                    __func__, info->host_model, info->host);
        free_hostnode(host);
        return NULL;
    }

    if (NULL == info->host_type) {
        host->type_index = 0;
    } else if ((host->type_index = get_hosttype_index(info->host_type)) < 0) {
        log_message(ERR, NOCLASS, "%s: Unknown host type %s for host <%s>.",
                    __func__, info->host_type, info->host);
        free_hostnode(host);
        return NULL;
    }

    host->name = copy_string(info->host);
    host->index = HASH_NUM_ENTS(gptrMyCluster->node_table);
    host->max_jobs = info->max_jobs;

    unsigned int i;
    for (i = 0; i < gptrSharedConf->qpinfo->num_index; i++) {
        host->busy_threshold[i] = info->load_threshold[i];
    }

    if (NULL == host->run_winlist) {
        host->run_winlist = gf_list_make("node run window list");
    }

    if (NULL != info->run_window && '\0' != info->run_window[0]) {
        parse_time_window(info->run_window, host->run_winlist, "host");
    }
    host->winclose_time = time(NULL);

    host->server = info->server;
    if (host->server == CLIENT) {
        host->noload_count = -1;
        gptrMyCluster->num_clients ++;
    }

    gf_hash_install(gptrMyCluster->node_table, host->name, (void *)host);

    return host;
} // end function hostinfo_2node

static void
free_res_instances(void *inst)
{
    struct resourceInstance *instance = (struct resourceInstance*)inst;

    if (NULL == instance) {
        return;
    }

    gf_hash_free(instance->node_table, gf_free_none);
    FREEUP(instance->value);
    FREEUP(instance->org_value);
    FREEUP(instance);

    return;
} // end function free_res_instances

/**
 * @brief       Parse node in LOCATION of ResMap section.
 *
 * @param[in]   hostlist     #1: node string inside []
 * @param[out]  hosts        #2: string array of node
 * @param[out]  flag         #3: flag to indicate if default or all insied []
 * @param[in]   resname      #4: resource name
 *
 * @retval      >0           #1: succeed, number of node in []
 * @retval      <0           #3: failed, no node defined in [] or system call failure
 * 
 * @note        'all', 'default', host type or host model are valid.
 */
static int
parse_nodelist(char *hostlist, char ***hosts, int *flag, char *resname)
{
    if (hostlist == NULL || hosts == NULL) {
        return -1;
    }

    int numHosts = 0;
    char *sp;
    sp = hostlist;
    unsigned int len = strlen(sp);

    char *host = (char *)calloc(len+1, sizeof(char));
    if (NULL == host) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc", len+1);
        jm_die(DAEMON_QUIT_MEM);
    }

    char *token;
    while ((token = get_string_token(&sp, " ", host, len+1)) != NULL) {
        len = strlen(sp);
        numHosts++;
    }

    if (0 == numHosts) {
        log_message(ERR, NOCLASS, "%s: no valid node specified in resource <%s> location <%s>.",
                    __func__, resname, hostlist);
        FREEUP(host);

        return -1;
    }

    *hosts = (char **)calloc(numHosts, sizeof(char *));
    if (NULL == *hosts) {

        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", numHosts*sizeof(char *));
        FREEUP(host);

        jm_die(DAEMON_QUIT_MEM);
    }

    sp = hostlist;
    len = strlen(sp);
    numHosts = 0;
    while ((token = get_string_token(&sp, " ", host, len+1)) != NULL) {
        len = strlen(sp);

        if (0 == strcmp(token, DEFAULT_STRING)) {

            *flag |= RESOURCE_FLAG_DEFAULT_ALL;

            if (0 < numHosts) {
                log_message(ERR, NOCLASS, "%s: keyword 'default' is specified for resource <%s> location <%s>, other nodes and instances are ignored.",
                            __func__, resname, hostlist);

                int i;
                for (i=0; i<numHosts; i++) {
                    FREEUP((*hosts)[i]);
                }

                numHosts = 0;
            }

            (*hosts)[numHosts] = copy_string(DEFAULT_STRING);
            numHosts++;

            continue;
        }

        if (0 == strcmp(token, ALL_STRING)) {

            *flag |= RESOURCE_FLAG_SHARED_ALL;

            if (numHosts > 0) {
                log_message(ERR, NOCLASS, "%s: keyword 'all' is specified for resource <%s> location <%s>, other nodes and instances are ignored.",
                            __func__, resname, hostlist);

                int i;
                for (i=0; i<numHosts; i++) {
                    FREEUP((*hosts)[i]);
                }

                numHosts = 0;
            }

            (*hosts)[numHosts] = copy_string(ALL_STRING);
            numHosts++;

            continue;
        }

        if (NULL != gf_hash_find(gptrMyCluster->node_table, token)) {

            (*hosts)[numHosts] = copy_string(token);
            numHosts++;

            continue;
        }

        if (get_hosttype_index(token) > 0) {
            (*hosts)[numHosts] = copy_string(token);
            numHosts++;

            continue;
        }

        if (get_hostmodel_index(token) > 0) {
            (*hosts)[numHosts] = copy_string(token);
            numHosts++;

            continue;
        }

        log_message(ERR, NOCLASS, "%s: cannot recognize the node[%s] specified in resource <%s> location <%s>.",
                    __func__, token, resname, hostlist);
    }
    FREEUP(host);

    if (0 == numHosts) {
        log_message(ERR, NOCLASS, "%s: all nodes specified in resource <%s> location <%s> are unknown.",
                    __func__, resname, hostlist);
        FREEUP(*hosts);

        return -1;
    }

    return numHosts;
} // end function parse_nodelist

/**
 * @brief       Add a share resource in the global hash table ghashResource.
 *
 * @param[in]   res          #1: share resource
 * @param[in]   numhosts     #2: number of node in []
 * @param[in]   hosts        #3: nodes in []
 * @param[in]   value        #4: static value of the resource
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, invalid input argument
 * 
 * @note        Add one node instance for shared all resource.
 *              Add multiple node instances for the same type of hosts or the same model of hosts.
 */
static int
add_resource_instance(struct sharedResource *res, unsigned int numhosts, char **host, char *value)
{
    if (NULL == host || NULL == res) {
        return -1;
    }

    struct resourceInstance *resInst;
    resInst = (struct resourceInstance *)calloc(1, sizeof(struct resourceInstance));
    if (NULL == resInst) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", sizeof(struct resourceInstance));
        jm_die(DAEMON_QUIT_MEM);
    }

    resInst->resource = res->resource;

    if (NULL == res->inst_list) {
        res->inst_list = gf_list_make("resource instance list");
    }

    resInst->node_table = gf_hash_make(numhosts);

    unsigned int i;
    for (i=0; i<numhosts; i++) {

        if (host[i] == NULL) {
            continue;
        }

        struct hostNode *node;
        node = (struct hostNode *)gf_hash_find(gptrMyCluster->node_table, host[i]);
        if (NULL != node) {

            gf_hash_install(resInst->node_table, host[i], (void *)node);
            if (NULL == node->inst_table) {
                node->inst_table = gf_hash_make(gptrSharedConf->qpinfo->num_resources);
            }
            gf_hash_install(node->inst_table, resInst->resource->name, (void *)resInst);  
            if (i == 0) {
                resInst->update_host = node;
            }

            if (NULL == node->res_bitmap) {
                node->res_bitmap = (int *)calloc(GET_INTNUM(gptrSharedConf->qpinfo->num_resources),
                                                 sizeof(int));
                if (NULL == node->res_bitmap) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc",
                                GET_INTNUM(gptrSharedConf->qpinfo->num_resources)*sizeof(int));
                    jm_die(DAEMON_QUIT_MEM);
                }
            }

            SET_BIT(res->index, node->res_bitmap);

            continue;
        }

        int num;

        if (0 == strcmp(host[i], ALL_STRING)) {
            node = (struct hostNode *)calloc(1, sizeof(struct hostNode));
            if (NULL == node) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                            "calloc", sizeof(struct hostNode));
                jm_die(DAEMON_QUIT_MEM);
            }

            node->name = copy_string(host[i]);

            gf_hash_install(resInst->node_table, node->name, (void *)node);

            HASH_WALK_T walk;
            char *key;
            gf_hash_walk_start(gptrMyCluster->node_table, &walk);
            struct hostNode *node4All;
            while (NULL != (node4All = (struct hostNode *)gf_hash_walk(&walk, &key))) {
                if (NULL == node4All->inst_table) {
                    node4All->inst_table = gf_hash_make(gptrSharedConf->qpinfo->num_resources);
                }
                gf_hash_install(node4All->inst_table, resInst->resource->name, (void *)resInst);
            }
            gf_hash_walk_end(&walk);

            continue;
        } else if ((num = get_hosttype_index(host[i])) > 0) {
            node = (struct hostNode *)calloc(1, sizeof(struct hostNode));
            if (NULL == node) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                            "calloc", sizeof(struct hostNode));
                jm_die(DAEMON_QUIT_MEM);
            }

            node->name = copy_string(host[i]);

            gf_hash_install(resInst->node_table, node->name, (void *)node);

            HASH_WALK_T walk;
            char *key;
            gf_hash_walk_start(gptrMyCluster->node_table, &walk);
            struct hostNode *node4Type;
            while (NULL != (node4Type = (struct hostNode *)gf_hash_walk(&walk, &key))) {
                if (node4Type->type_index != num) {
                    continue;
                }

                if (NULL == node4Type->inst_table) {
                    node4Type->inst_table = gf_hash_make(gptrSharedConf->qpinfo->num_resources);
                }
                gf_hash_install(node4Type->inst_table, resInst->resource->name, (void *)resInst);

                if (NULL == node4Type->res_bitmap) {
                    node4Type->res_bitmap = (int *)calloc(GET_INTNUM(gptrSharedConf->qpinfo->num_resources),
                                                          sizeof(int));
                    if (NULL == node4Type->res_bitmap) {
                        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc",
                                    GET_INTNUM(gptrSharedConf->qpinfo->num_resources)*sizeof(int));
                        jm_die(DAEMON_QUIT_MEM);
                    }
                }

                SET_BIT(res->index, node4Type->res_bitmap);
            }
            gf_hash_walk_end(&walk);

            continue;
        } else if ((num = get_hostmodel_index(host[i])) > 0) {
            node = (struct hostNode *)calloc(1, sizeof(struct hostNode));
            if (NULL == node) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                            "calloc", sizeof(struct hostNode));
                jm_die(DAEMON_QUIT_MEM);
            }

            node->name = copy_string(host[i]);

            gf_hash_install(resInst->node_table, node->name, (void *)node);

            HASH_WALK_T walk;
            char *key;
            gf_hash_walk_start(gptrMyCluster->node_table, &walk);
            struct hostNode *node4Type;
            while (NULL != (node4Type = (struct hostNode *)gf_hash_walk(&walk, &key))) {
                if (node4Type->model_index != num) {
                    continue;
                }

                if (NULL == node4Type->inst_table) {
                    node4Type->inst_table = gf_hash_make(gptrSharedConf->qpinfo->num_resources);
                }
                gf_hash_install(node4Type->inst_table, resInst->resource->name, (void *)resInst);

                if (NULL == node4Type->res_bitmap) {
                    node4Type->res_bitmap = (int *)calloc(GET_INTNUM(gptrSharedConf->qpinfo->num_resources),
                                                          sizeof(int));
                    if (NULL == node4Type->res_bitmap) {
                        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc",
                                    GET_INTNUM(gptrSharedConf->qpinfo->num_resources)*sizeof(int));
                        jm_die(DAEMON_QUIT_MEM);
                    }
                }

                SET_BIT(res->index, node4Type->res_bitmap);
            }
            gf_hash_walk_end(&walk);

            continue;
        } else {
            log_message(ERR, NOCLASS, "%s: node <%s> is unknown to me.", __func__, host[i]);
        }
    }

    resInst->flags = res->flags;

    if (NULL == value || '\0' == value[0]) {
        resInst->value = copy_string((char *)"-");
        resInst->org_value = copy_string((char *)"-");
    } else {
        resInst->value = copy_string(value);
        resInst->org_value = copy_string(value);
    }

    gf_list_insert(res->inst_list, res->inst_list->forw, (LIST_T *)resInst);

    return 0;
} // end function add_resource_instance

/**
 * @brief       Add a share resource in the global hash table ghashResource.
 *
 * @param[in]   resname      #1: name of the share resource
 * @param[in]   index        #2: resource index
 * @param[in]   numhosts     #3: number of node in []
 * @param[in]   hosts        #4: nodes in []
 * @param[in]   value        #5: static value of the resource
 *
 * @retval      not NULL     #1: succeed
 * @retval      NULL         #2: failed, system call failure
 * 
 * @note        For location[all], we add one node instance to the share resource.
 *              For location[default], we add instance to every node in my cluster.
 *              Dynamic resource does not have value defined.
 */
static struct sharedResource *
add_shared_resource(char *resname, int index, int numhosts, char **hosts, char *value)
{
    if (NULL == resname || hosts == NULL) {
        return NULL;
    }

    struct sharedResource *res;
    res = (struct sharedResource *)calloc(1, sizeof(struct sharedResource));
    if (NULL == res) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", sizeof(struct sharedResource));
        jm_die(DAEMON_QUIT_MEM);
    }

    res->resource = &(gptrSharedConf->qpinfo->resource_table[index]);
    res->index = index;

    if (numhosts == 1 && strcmp(hosts[0], ALL_STRING) == 0) {

        res->flags |= RESOURCE_FLAG_SHARED_ALL;
        add_resource_instance(res, numhosts, hosts, value);
    } else if (numhosts == 1 && strcmp(hosts[0], "default") == 0) {
        res->flags |= RESOURCE_FLAG_DEFAULT_ALL;

        HASH_WALK_T walkNode;
        char *key;
        struct hostNode *host;

        gf_hash_walk_start(gptrMyCluster->node_table, &walkNode);
        while (NULL != (host=(struct hostNode *)gf_hash_walk(&walkNode, &key))) {
            if (CLIENT == host->server) {
                continue;
            }

            add_resource_instance(res, 1, &(host->name), value);
        }
        gf_hash_walk_end(&walkNode);
    } else {
        add_resource_instance(res, numhosts, hosts, value);
    }

    if (NULL == ghashResource) {
        ghashResource = gf_hash_make(23);
    }

    gf_hash_install(ghashResource, resname, (void *)res);

    return res;
} // end function add_shared_resource

/**
 * @brief       Transfer location to share resource instance.
 *
 * @param[in]   resloc       #1: resource location read from configuration file
 *
 * @note        Parse resource location and add to share resource data accordingly.
 */
static void
location_2resinst(struct resourceLoc *resloc)
{
    int resIndex;

    if ((resIndex = get_resource_index_fromhash(resloc->resource_name)) < 0) {
        log_message(ERR, NOCLASS, "%s: resource <%s> is not defined.",
                    __func__, resloc->resource_name);
        return;
    }

    struct sharedResource *resource;
    resource = (struct sharedResource *)gf_hash_find(ghashResource, resloc->resource_name);

    int flag=0;
    char *sp = resloc->location;
    while (sp != NULL && sp[0] != '\0') {
        char *cp = sp;
        if (*cp != '[' && *cp != '\0') {
            while (*cp && *cp != '@' && !(!iscntrl(*cp) && isspace(*cp))) {
                cp++;
            }
        }

        char *strLocValue=NULL;
        if (cp != sp) {
            char ssp = cp[0];

            cp[0] = '\0';
            strLocValue = copy_string(sp);
            cp[0] = ssp;

            while (isspace(*cp) && *cp != '\0') {
                cp++;
            }

            sp = cp + 1;
        }

        while (isspace(*sp) && *sp != '\0') {
            sp++;
        }

        char *strNode = get_string_inside(&sp, '[', ']');
        cp = strip_spaces(strNode, FALSE);

        int numHosts;
        char **hosts=NULL;
        if ((numHosts = parse_nodelist(cp, &hosts, &flag, resloc->resource_name)) <= 0) {

            FREEUP(strNode);

            continue;
        }
        FREEUP(strNode);

        unsigned int numInst = (NULL==resource)?0:LIST_NUM_ENTS(resource->inst_list);
        if (((flag & RESOURCE_FLAG_SHARED_ALL) || (flag & RESOURCE_FLAG_DEFAULT_ALL))
            && (0 < numInst)) {

            log_message(ERR, NOCLASS, "%s: keyword 'all' or 'default' is specified for resource <%s> instance, other instances are ignored.",
                        __func__, resloc->resource_name);

            if (NULL != resource) {
                gf_list_free(resource->inst_list, free_res_instances);

                resource->inst_list = gf_list_make("resource instance list");
            }
        }

        if (NULL == resource) {
            resource = add_shared_resource(resloc->resource_name, resIndex,
                                           numHosts, hosts, strLocValue);
        } else {
            add_resource_instance(resource, numHosts, hosts, strLocValue);
        }

        int i;
        for (i=0; i<numHosts; i++) {
            FREEUP(hosts[i]);
        }
        FREEUP(hosts);
        FREEUP(strLocValue);

        if (!(flag & RESOURCE_FLAG_DEFAULT_ALL
              && (gptrSharedConf->qpinfo->resource_table[resIndex].flags & RESOURCE_FLAG_DYNAMIC)
              && (gptrSharedConf->qpinfo->resource_table[resIndex].value_type == TYPE_NUMERIC))) {

            gptrSharedConf->qpinfo->resource_table[resIndex].flags &= ~RESOURCE_FLAG_GLOBAL;
            gptrSharedConf->qpinfo->resource_table[resIndex].flags |= RESOURCE_FLAG_SHARED;

            if (gptrSharedConf->qpinfo->resource_table[resIndex].flags & RESOURCE_FLAG_DYNAMIC
                && gptrSharedConf->qpinfo->resource_table[resIndex].value_type == TYPE_NUMERIC) {

                gptrSharedConf->qpinfo->num_user_index--;
                gptrSharedConf->qpinfo->num_index--;
            }
        }

        if (gptrSharedConf->qpinfo->resource_table[resIndex].flags
            & (RESOURCE_FLAG_DYNAMIC|RESOURCE_FLAG_GLOBAL)) {

            if (flag & RESOURCE_FLAG_DEFAULT_ALL) {
                gboolDefaultUnrm = TRUE;
            }
        }
    }

    return;
} // end function location_2resinst

/**
 * @brief       Reorder the resources in resource table.
 *
 * @note        The oder of resources is:
 *                 Dynamic shared numeric resource (include built-in resources)
 *                 Static numeric resource
 *                 Boolean resource
 *                 String resource
 */
static void
reorder_resource(void)
{
    struct resource_definition *newTable;

    gptrSharedConf->qpinfo->num_index = 0;
    gptrSharedConf->qpinfo->num_user_index = 0;
    newTable = (struct resource_definition *)calloc(gptrSharedConf->qpinfo->num_resources,
                                                    sizeof(struct resource_definition));
    if (NULL == newTable) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__, "calloc",
                    gptrSharedConf->qpinfo->num_resources*sizeof(struct resource_definition));
        jm_die(DAEMON_QUIT_MEM);
    }

    unsigned int i, j;
    for (i = 0, j = 0; i < gptrSharedConf->qpinfo->num_resources; i++) {

        if (gptrSharedConf->qpinfo->resource_table[i].value_type == TYPE_NUMERIC
            && (gptrSharedConf->qpinfo->resource_table[i].flags & RESOURCE_FLAG_DYNAMIC)
            && (gptrSharedConf->qpinfo->resource_table[i].flags & RESOURCE_FLAG_GLOBAL)) {

            memcpy(&newTable[j], &gptrSharedConf->qpinfo->resource_table[i],
                   sizeof(struct resource_definition));

            j++;
        }
    }

    for (i = 0; i < gptrSharedConf->qpinfo->num_resources; i++) {

        if (gptrSharedConf->qpinfo->resource_table[i].value_type == TYPE_NUMERIC
            && (!(gptrSharedConf->qpinfo->resource_table[i].flags & RESOURCE_FLAG_DYNAMIC)
                || !(gptrSharedConf->qpinfo->resource_table[i].flags & RESOURCE_FLAG_GLOBAL))) {

            memcpy(&newTable[j], &gptrSharedConf->qpinfo->resource_table[i],
                   sizeof(struct resource_definition));

            j++;
        }
    }

    bool_t start=FALSE;
    for (i = 0; i < gptrSharedConf->qpinfo->num_resources; i++) {
        if (gptrSharedConf->qpinfo->resource_table[i].value_type == TYPE_BOOLEAN) {

            memcpy(&newTable[j], &gptrSharedConf->qpinfo->resource_table[i],
                   sizeof(struct resource_definition));

            if (!start) {
                start = TRUE;
                gptrSharedConf->qpinfo->bool_start_index = j;
            }

            j++;
        }
    }

    start = FALSE;
    for (i = 0; i < gptrSharedConf->qpinfo->num_resources; i++) {
        if (gptrSharedConf->qpinfo->resource_table[i].value_type == TYPE_STRING) {

            memcpy(&newTable[j], &gptrSharedConf->qpinfo->resource_table[i],
                   sizeof(struct resource_definition));

            if (!start) {
                start = TRUE;
                gptrSharedConf->qpinfo->string_start_index = j;
            }

            j++;
        }
    }

    free(gptrSharedConf->qpinfo->resource_table);
    gptrSharedConf->qpinfo->resource_table = newTable;

    unsigned int numRes = gptrSharedConf->qpinfo->num_resources;
    gptrMyCluster->num_numeric_resources = 0;
    gptrMyCluster->numeric_resource_names = (char **)calloc(numRes, sizeof(char *));
    if (NULL == gptrMyCluster->numeric_resource_names) {

        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", numRes*sizeof(char *));
        jm_die(DAEMON_QUIT_MEM);
    }

    numRes = 0;
    for (i = 0; i < gptrSharedConf->qpinfo->num_resources; i++) {

        if ((gptrSharedConf->qpinfo->resource_table[i].flags & RESOURCE_FLAG_DYNAMIC)
            && (gptrSharedConf->qpinfo->resource_table[i].value_type == TYPE_NUMERIC)
            && (gptrSharedConf->qpinfo->resource_table[i].flags & RESOURCE_FLAG_GLOBAL)) {

            if (!(gptrSharedConf->qpinfo->resource_table[i].flags & RESOURCE_FLAG_BUILTIN)) {
                gptrSharedConf->qpinfo->num_user_index++;
            }
            gptrSharedConf->qpinfo->num_index++;
        }

        if ((gptrSharedConf->qpinfo->resource_table[i].flags & RESOURCE_FLAG_BUILTIN)
            || (gptrSharedConf->qpinfo->resource_table[i].flags & RESOURCE_FLAG_DYNAMIC
                && gptrSharedConf->qpinfo->resource_table[i].value_type == TYPE_NUMERIC)
            || (gptrSharedConf->qpinfo->resource_table[i].value_type != TYPE_STRING
                && gptrSharedConf->qpinfo->resource_table[i].value_type != TYPE_BOOLEAN)) {
            continue;
        }

        char *resName = gptrSharedConf->qpinfo->resource_table[i].name;
        gptrMyCluster->numeric_resource_names[numRes] = copy_string(resName);
        numRes ++;

    }
    gptrMyCluster->num_numeric_resources = numRes;

    return;
} // end function reorder_resource

/**
 * @brief       popen function for starting unrm script.
 *
 * @param[in]   argv         #1: command and argument of unrm
 * @param[in]   mode         #2: only support 'r' for now
 *
 * @retval      not NULL     #1: succeed, pipe handler
 * @retval      NULL         #2: failed, system call failure
 *
 * @note        Fork a child and execute the unrm script.
 */
static FILE *
unrm_popen(char **argv, const char *mode)
{
    int p[2], i;

    if (mode[0] != 'r') {
        return NULL;
    }

    if (0 > pipe(p)) {
        return NULL;
    }

    pid_t pid;
    pid = fork();
    if (0 > pid) {
        close(p[0]);
        close(p[1]);
        return NULL;
    }

    if (0 == pid) {

        close(p[0]);
        dup2(p[1], 1);

        alarm(0);

        for(i = 2; i < sysconf(_SC_OPEN_MAX); i++) {
            close(i);
        }

        for (i = 1; i < NSIG; i++) {
            Signal_(i, SIG_DFL);
        }

        execvp(argv[0], argv);
        log_message(ERR, NOCLASS, STR_FUNC_S_FAIL_M, __func__, "execvp", argv[0]);
        exit(127);
    }

    gpidUnrm = pid; // set unrm pid
    close(p[1]);

    return fdopen(p[0], mode);
} // end function unrm_popen

/**
 * @brief       pclose function for ending unrm script.
 *
 * @param[in]   pipe         #1: pipe to be closed
 *
 * @retval      0            #1: succeed
 * @retval      -1           #2: failed, system call failure
 */
static int
unrm_pclose(FILE *pipe)
{
    sigset_t   omask;
    sigset_t   newmask;
    pid_t      child;

    child = gpidUnrm;
    gpidUnrm = -1;

    if (NULL != pipe) {
        fclose(pipe);
    }

    if (0 > child) {
        return -1;
    }

    kill(child, SIGTERM);

    sigemptyset(&newmask);
    sigaddset(&newmask, SIGINT);
    sigaddset(&newmask, SIGQUIT);
    sigaddset(&newmask, SIGHUP);
    sigprocmask(SIG_BLOCK, &newmask, &omask);

    sigprocmask(SIG_SETMASK, &omask, NULL);

    return 0;
} // end function unrm_pclose

/**
 * @brief       kill the unrm process.
 */
static void
terminate_unrm(void)
{
    if (0 > gpidUnrm) {
        return;
    }

    kill(gpidUnrm, SIGTERM);
    gpidUnrm = -1;

    return;
} // end function terminate_unrm

/**
 * @brief       check if I should start unrm or not.
 *
 * @retval      TRUE         #1: to start unrm script
 * @retval      FALSE        #2: do not run unrm script
 * 
 * @note        For master node, should start unrm for shared all resource
 *              For share resource, start one unrm per location
 *              If has location[default] resource, should start unrm on each node
 */
static int
call_unrm(void)
{
    static int runit = FALSE;
    static int lastTimeMasterMe = FALSE;

    if (gboolMaster && !lastTimeMasterMe) {
        lastTimeMasterMe = TRUE;
        if (runit){
            terminate_unrm();
            if (gptrMyHost->call_unrm || gboolDefaultUnrm)
                return TRUE;

            runit = FALSE;
            return FALSE;
        }
    }

    if (!gboolMaster && lastTimeMasterMe){
        lastTimeMasterMe = FALSE;

        if (runit){
            terminate_unrm();

            if (gptrMyHost->call_unrm || gboolDefaultUnrm)
                return TRUE;

            runit = FALSE;
            return FALSE;
        }
    }

    if (gboolMaster)
        lastTimeMasterMe = TRUE;
    else
        lastTimeMasterMe = FALSE;

    if (runit) {
        if (!gptrMyHost->call_unrm && !gboolDefaultUnrm){
            terminate_unrm();
            runit = FALSE;
            return FALSE;
        }
    }

    if (gboolDefaultUnrm) {
        runit = TRUE;
        return TRUE;
    }

    if (gptrMyHost->call_unrm) {
        runit = TRUE;
        return TRUE;
    }
    runit = FALSE;

    return FALSE;
} // end function call_unrm

static void
set_unknown_values(void)
{
    unsigned int i;
    for (i=0; i<gptrSharedConf->qpinfo->num_user_index; i++) {
        gptrMyHost->load[NUMBUILTIN + i] = INFINIT_LOAD;
    }

    HASH_WALK_T walk;
    char *key;
    struct resourceInstance *inst;
    gf_hash_walk_start(gptrMyHost->inst_table, &walk);
    while (NULL != (inst=(struct resourceInstance *)gf_hash_walk(&walk, &key))) {
        if (0 == inst->update_time || NULL == inst->update_host) {
            continue;
        }

        if (inst->update_host == gptrMyHost) {
            strcpy(inst->value, "-");
            inst->update_time = 0;
        }
    }
    gf_hash_walk_end(&walk);

    return;
} // end function set_unknown_values

static int
set_dynamicshared_value(char *name, char *value)
{
    int indx;

    if ((indx = get_resource_index_fromhash(name)) < 0) {
        return -1;
    }

    if (!(gptrSharedConf->qpinfo->resource_table[indx].flags & RESOURCE_FLAG_DYNAMIC)) {
        return -1;
    }

    if (gptrSharedConf->qpinfo->resource_table[indx].value_type == TYPE_NUMERIC) {
        if (!isnumber(value)){
            return -1;
        }
    }

    struct resourceInstance *inst;
    inst = (struct resourceInstance *)gf_hash_find(gptrMyHost->inst_table, name);
    if (NULL == inst) {
        log_message(ERR, NOCLASS, "%s: Jm can collect share resource[%s] value[%s], but can't match the name in resource table",
                __func__, name, value);        
        return -1;
    }

    if (inst->update_host != gptrMyHost) {
        return 0;
    }

    FREEUP(inst->org_value);
    if (NULL != inst->value) {
        inst->org_value = copy_string(inst->value);
    }

    FREEUP(inst->value);
    inst->value = copy_string(value);
    inst->update_time = time(NULL);

    log_message(DEBUG, DEBUGRES, "%s: resname %s value %s host %s",
                __func__, inst->resource->name, inst->value, inst->update_host->name);

    return 0;
} // end function set_dynamicshared_value

/**
 * @brief       Get output of unrm and set resource values.
 *
 * @note        Use pipe for communication.
 */
static void
get_unrm_report(void)
{
    if (!call_unrm()) {
        return;
    }

    if ((0 > gpidUnrm) && (time(NULL) - stimeCheckUnrm > 90)) {

        if (NULL == gptrMyCluster->unrm_argv) {

            char path[PATH_MAX];
            strcpy(path, baseParams[BASE_SERVERDIR].string_value);
            strcat(path, "/");
            strcat(path, "unrm");
    
            struct stat sbuf;
            if (stat(path, &sbuf) < 0) {

                if (errno == ENOENT) {
                    return;
                }

                log_message(ERR, NOCLASS, STR_FUNC_FAIL_M, __func__, "stat");
                return;
            }

            gptrMyCluster->unrm_argv = (char **)calloc(2, sizeof(char *));
            if (NULL == gptrMyCluster->unrm_argv) {

                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                            "calloc", 2*sizeof(char *));
                jm_die(DAEMON_QUIT_MEM);
            }

            gptrMyCluster->unrm_argv[0] = copy_string(path);
        }

        if (NULL != spipeUnrm) {
            fclose(spipeUnrm);
        }

        stimeCheckUnrm = time(NULL);

        if (gboolMaster) {
            gf_setenv("QP_MASTER", (char *)"Y");
        } else {
            gf_setenv("QP_MASTER", (char *)"N");
        }

        spipeUnrm = unrm_popen(gptrMyCluster->unrm_argv, "r");
        if (NULL == spipeUnrm) {
            log_message(ERR, NOCLASS, "%s: unrm_popen(%s) failed due to %m.",
                        __func__, gptrMyCluster->unrm_argv[0]);
            set_unknown_values();

            return;
        }

        log_message(INFO, NOCLASS, "%s: PID of started UNRM <%s> is %d.",
                    __func__, gptrMyCluster->unrm_argv[0], gpidUnrm);
    }

    if (0 > gpidUnrm) {

        set_unknown_values();
        if (NULL != spipeUnrm) {
            fclose(spipeUnrm);
            spipeUnrm = NULL;
        }

        return;
    }

    struct timeval timeout;
    timeout.tv_sec  = 0;
    timeout.tv_usec = 5;

    int nfds;
    if ((nfds = gf_read_select(fileno(spipeUnrm), timeout)) < 0) {
        log_message(ERR, NOCLASS, "%s: rd_select() failed %m", __func__);
        unrm_pclose(spipeUnrm);
        spipeUnrm = NULL;
        return;
    }

    if (1 == nfds) {
        int cc;
        static char name[MAXNAMELEN];
        static char svalue[MAXNAMELEN];
        float value;
        sigset_t oldMask;
        sigset_t newMask;

        gf_block_signals(0, &newMask, &oldMask);

        log_message(DEBUG, DEBUGSIGNAL, "%s: Signal mask has been changed, all signals are blocked now", __func__);

        unsigned int numIndx;
        cc = fscanf(spipeUnrm, "%d", &numIndx);
        if (1 != cc) {
            log_message(ERR, NOCLASS, "%s: Protocol error numIndx not read (cc=%d): %m", __func__, cc);
            unrm_pclose(spipeUnrm);
            spipeUnrm = NULL;
            gf_unblock_signals(&oldMask);
            return;
        }

        if (0 > numIndx) {
            log_message(ERR, NOCLASS, "%s: protocol error, numIndx <%d> is less than 0.",
                        __func__, numIndx);
            set_unknown_values();
            unrm_pclose(spipeUnrm);
            spipeUnrm = NULL;
            gf_unblock_signals(&oldMask);
            return;
        }

        unsigned int i;
        for (i = 0; i < numIndx; i++) {
            cc = fscanf(spipeUnrm,"%s %s", name, svalue);
            if (cc != 2) {
                log_message(ERR, NOCLASS, "%s: Protocol error on indx %d (cc = %d): %m", __func__, i, cc);
                unrm_pclose(spipeUnrm);
                spipeUnrm = NULL;
                return;
            }

            /* Shared resource. */
            if (set_dynamicshared_value(name, svalue) == 0) {
                continue;
            }

            value = atof(svalue);

            /* Load index. */
            int indx;
            indx = get_resource_index_fromhash(name);
            if (indx < 0) {
                log_message(ERR, NOCLASS, "%s: Unknown index name %s from UNRM", __func__, name);
                continue;
            }

            if (gptrSharedConf->qpinfo->resource_table[indx].value_type != TYPE_NUMERIC
                || (unsigned int)indx >= gptrSharedConf->qpinfo->num_index) {
                continue;
            }

            gptrMyHost->load[indx] = value;
        }

        gf_unblock_signals(&oldMask);
    } /* if (nfds == 1) */

    return;
} // end function get_unrm_report

#if defined (HAVE_HWLOC_H)
/**
 * @brief       set numa child object.
 *
 * @param[in]       parent_obj   #1: parent object
 * @param[in]       parent_node  #2: parent numa object
 * @param[in]       type         #3: NUMA_CORE
 * @param[in|out]   index        #4: core index
 *
 * @retval          >=0          #1: succeed
 */
static int  
set_numa_child(hwloc_obj_t parent_obj, numa_obj_t *parent_node, int type, int *index)
{
    int num_core = 0;
    numa_obj_t *last_node=NULL;
    hwloc_obj_t obj_tmp;
    numa_obj_t *node = NULL;

    obj_tmp = parent_obj->first_child;
    unsigned int i;
    for (i=0; i<obj_tmp->arity; i++) {

        hwloc_obj_t objCore = obj_tmp->children[i];
        while (objCore != NULL) {
            if (type == NUMA_CORE && objCore->type != HWLOC_OBJ_CORE) {
                objCore = objCore->first_child;
            } else {
                break;
            }
        }

        node = (numa_obj_t *)calloc(1, sizeof(struct numa_obj));  
        if (NULL == node) {
            log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                        "calloc", sizeof(struct numa_obj));
            free_numanode(sptrMyNuma);
            jm_die(DAEMON_QUIT_MEM);
        } 

        if (parent_node->child == NULL) {
            parent_node->child = node;
            last_node = node;  
        } else {
            last_node->back = node;
            node->forw = last_node;
            last_node = node;
        }        

        node->parent = parent_node;
        node->type = type;
        node->index_sibling = objCore->os_index;
        node->index = (*index)++;
        num_core++;
    }

    return num_core;
} // end function set_numa_child

/**
 * @brief       Get node numa topology.
 *
 * @note        call hwloc functions to get node numa information.
 */
static void
get_numa_info(void)
{    
    /* Allocate and initialize topology object. */
    if (hwloc_topology_init(&topology) == -1) {
        log_message(ERR, NOCLASS, "%s: Hwloc topology init failed.", __func__);
        return;
    }
    /* Perform the topology detection. */
    if (hwloc_topology_load(topology) == -1) {
        log_message(ERR, NOCLASS, "%s: Hwloc topology load failed.", __func__);
        return;
    }
    hwloc_obj_t obj, obj_socket, obj_machine;
    numa_obj_t *virtual_numanode = NULL, *numa_lastnode = NULL, *numa_lastsocket = NULL;
    obj_machine = hwloc_get_obj_by_type(topology, HWLOC_OBJ_MACHINE, 0);
    sptrMyNuma = (numa_obj_t *)calloc(1, sizeof(struct numa_obj));
    if (NULL == sptrMyNuma) {
        log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                    "calloc", sizeof(struct numa_obj));
        jm_die(DAEMON_QUIT_MEM);
    }

    sptrMyNuma->type = NUMA_HOST;        
    
    //unit is M
    sptrMyNuma->memory = obj_machine->memory.total_memory / 1024 / 1024;
    
    int node_index = 0, socket_index = 0, core_index = 0;
    for (obj=obj_machine->first_child; obj != NULL; obj = obj->next_sibling) {
        if (obj->type == HWLOC_OBJ_NUMANODE) {
            numa_obj_t *numanode_node = (numa_obj_t *)calloc(1, sizeof(struct numa_obj));
            if (NULL == numanode_node) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                            "calloc", sizeof(struct numa_obj));
                free_numanode(sptrMyNuma);
                jm_die(DAEMON_QUIT_MEM);
                return;
            }            

            if (sptrMyNuma->child == NULL) {
                sptrMyNuma->child = numanode_node;
                numa_lastnode = numanode_node;  
                numanode_node->index = node_index++;
            } else {
                numa_lastnode->back = numanode_node;
                numanode_node->forw = numa_lastnode;
                numanode_node->index = node_index++;
                numa_lastnode = numanode_node;               
            }

            numanode_node->parent = sptrMyNuma;
            numanode_node->type = NUMA_NODE;
            numanode_node->index_sibling = obj->os_index;
            numanode_node->memory = obj->memory.total_memory / 1024 / 1024;

            obj_socket = obj->first_child;
            for (; obj_socket != NULL; obj_socket = obj_socket->next_sibling) {
                numa_obj_t *socket_node = (numa_obj_t *)calloc(1, sizeof(struct numa_obj));
                if (NULL == socket_node) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                                "calloc", sizeof(struct numa_obj));
                    free_numanode(sptrMyNuma);
                    jm_die(DAEMON_QUIT_MEM);
                    return;
                }            

                if (numanode_node->child == NULL) {
                    numanode_node->child = socket_node;
                    numa_lastsocket = socket_node;
                    socket_node->index = socket_index++;
                } else {
                    numa_lastsocket->back = socket_node;
                    socket_node->forw = numa_lastsocket;
                    numa_lastsocket = socket_node;
                    socket_node->index = socket_index++;
                }

                socket_node->parent = numanode_node;
                socket_node->type = NUMA_SOCKET;
                socket_node->index_sibling = obj_socket->os_index;    
                socket_node->child_num = set_numa_child(obj_socket, socket_node,
                                                        NUMA_CORE, &core_index);
                socket_node->total = socket_node->child_num;
                numanode_node->total += socket_node->total;
                numanode_node->child_num++;
            }
            sptrMyNuma->total += numanode_node->total;
            sptrMyNuma->child_num++;
        }
        if (obj->type == HWLOC_OBJ_PACKAGE) {
            if (NULL == virtual_numanode) {
                virtual_numanode = (numa_obj_t *)calloc(1, sizeof(struct numa_obj));
                if (NULL == virtual_numanode) {
                    log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                                "calloc", sizeof(struct numa_obj));
                    free_numanode(sptrMyNuma);
                    jm_die(DAEMON_QUIT_MEM);
                    return;
                }            
            }

            if (sptrMyNuma->child == NULL) {
                sptrMyNuma->child = virtual_numanode;   
                virtual_numanode->parent = sptrMyNuma;
                virtual_numanode->type = NUMA_NODE;
                virtual_numanode->index = 0;
                virtual_numanode->index_sibling = 0;
                virtual_numanode->memory = sptrMyNuma->memory;
                sptrMyNuma->child_num = 1;
            } 
        
            numa_obj_t *socket_node = (numa_obj_t *)calloc(1, sizeof(struct numa_obj));
            if (NULL == socket_node) {
                log_message(ERR, NOCLASS, STR_FUNC_D_FAIL_M, __func__,
                            "calloc", sizeof(struct numa_obj));
                free_numanode(sptrMyNuma);
                jm_die(DAEMON_QUIT_MEM);
                return;
            }            

            if (virtual_numanode->child == NULL) {
                virtual_numanode->child = socket_node;
                numa_lastsocket = socket_node;    
                socket_node->index = socket_index++;
            } else {
                numa_lastsocket->back = socket_node;
                socket_node->forw = numa_lastsocket;
                socket_node->index = socket_index++;
                numa_lastsocket = socket_node;
            }             
            socket_node->parent = virtual_numanode;
            socket_node->type = NUMA_SOCKET;
            socket_node->index_sibling = obj->os_index; 
            socket_node->child_num = set_numa_child(obj, socket_node, NUMA_CORE, &core_index);
            socket_node->total = socket_node->child_num;
            virtual_numanode->total += socket_node->total;
            virtual_numanode->child_num++;
            sptrMyNuma->total = virtual_numanode->total;
        }
    }   

    return;
} // end function get_numa_info
#endif
